<?php
/**
 * @Notes: 文件描述
 * @Interface getCondition
 * @Return mixed
 * @Author: bqs
 * @Time: 2021/4/19 15:28
 */

namespace app\common\command;

use think\console\Command;
use think\console\Input;
use think\console\Output;
use think\console\input\Argument;
use think\console\input\Option;
use think\Db;
use think\facade\Hook;
use think\facade\Log;

class KafkaCustomer extends Command
{

    protected function configure()
    {
        $this->setName('kafka:consumer')
            ->setDescription('产品同步到Es');
    }

    // protected function execute(Input $input, Output $output)
    // {
    //     $output->writeln("!!!hello kafka_customer!!!");
    //
    //     /**
    //      * See-kafka Consumer例子1
    //      * 889 毫秒 获取1w条
    //      */
    //     // 配置KafKa集群(默认端口9092)通过逗号分隔
    //     $KafKa_Lite = new KafKa_Lite("127.0.0.1,localhost");
    //
    //     // 设置一个Topic
    //     $KafKa_Lite->setTopic("test");
    //
    //     // 设置Consumer的Group分组(不使用自动offset的时候可以不设置)
    //     $KafKa_Lite->setGroup("test");
    //
    //     // 此项设置决定 在使用一个新的group时  是从 最小的一个开始 还是从最大的一个开始  默认是最大的(或尾部)
    //     $KafKa_Lite->setTopicConf('auto.offset.reset', 'smallest');
    //
    //     // 此项配置决定在获取数据后回自动作为一家消费 成功 无需在 一定要 stop之后才会 提交 但是也是有限制的
    //     // 时间越小提交的时间越快,时间越大提交的间隔也就越大 当获取一条数据之后就抛出异常时 更具获取之后的时间来计算是否算作处理完成
    //     // 时间小于这个时间时抛出异常 则不会更新offset 如果大于这个时间则会直接更新offset 建议设置为 100~1000之间
    //     $KafKa_Lite->setTopicConf('auto.commit.interval.ms', 1000);
    //
    //     // 获取Consumer实例
    //     $consumer = $KafKa_Lite->newConsumer();
    //
    //     // 开启Consumer获取,参数分别为partition(默认:0),offset(默认:KAFKA_OFFSET_STORED)
    //     $consumer->consumerStart(0);
    //
    //     for ($i = 0; $i < 100; $i++) {
    //         // 当获取不到数据时会阻塞默认10秒可以通过$consumer->setTimeout()进行设置
    //         // 阻塞后由数据能够获取会立即返回,超过10秒回返回null,正常返回格式为Kafka_Message
    //         $message = $consumer->consume();
    //     }
    //
    //     // 关闭Consumer(不关闭程序不会停止)
    //     $consumer->consumerStop();
    // }

    // // kafka低级消费者(只从指定的partition消费)
    protected function execute(Input $input, Output $output)
    {
        $output->writeln("!!!hello kafka!!!");

        $conf = new \RdKafka\Conf();
        $conf->set('group.id', 'sms-consumer-group');

        $rk = new \RdKafka\Consumer($conf);
        $rk->addBrokers("127.0.0.1:9092");

        $topicConf = new \RdKafka\TopicConf();
        $topicConf->set('auto.commit.interval.ms', 100);
        $topicConf->set('offset.store.method', 'file');
        $topicConf->set('offset.store.path', sys_get_temp_dir());
        $topicConf->set('auto.offset.reset', 'smallest');
        $topic = $rk->newTopic("shop", $topicConf);
        $topic->consumeStart(0, RD_KAFKA_OFFSET_STORED);

        while(true) {
            // 设置消费时的时间间隔，单位毫秒，以下表示5秒消费一个
            $message = $topic->consume(0, 5000);
            if ($message) {
                echo "读取到消息\n\r";

                // 消息对象，包括消息主题，消息创建时间戳，消息分区编号，消息主体，消息键名，消息长度等
                var_dump($message);

                switch ($message->err) {
                    case RD_KAFKA_RESP_ERR_NO_ERROR:
                        echo "读取消息成功:\n\r";
                        var_dump($message->payload);
                        break;
                    case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                        echo "读取消息失败\n\r";
                        break;
                    case RD_KAFKA_RESP_ERR__TIMED_OUT:
                        echo "请求超时\n\r";
                        break;
                    default:
                        throw new \Exception($message->errstr(), $message->err);
                        break;
                }
            } else {
                echo "未读取到消息\n\r";
            }
        }

        $output->writeln("!!!the end!!!");
    }

    // kafka低级消费者（动态自适应partition消费，也可以指定多个topic）
    // protected function execute(Input $input, Output $output)
    // {
    //     $output->writeln("!!!hello kafka!!!");
    //
    //     $conf = new \RdKafka\Conf();
    //
    //     $conf->setRebalanceCb(function (\RdKafka\KafkaConsumer $kafka, $err, array $partitions = null) {
    //         switch ($err) {
    //             case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
    //                 $kafka->assign($partitions);
    //                 break;
    //             case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
    //                 $kafka->assign(NULL);
    //                 break;
    //             default:
    //                 throw new \Exception($err);
    //         }
    //     });
    //
    //     $conf->set("group.id", "sms-consumer-group");
    //     $conf->set("metadata.broker.list", "127.0.0.1:9092");
    //
    //     // $topicConf = new \RdKafka\TopicConf();
    //     // $topicConf->set("auto.offset.reset", "smallest");
    //     // $conf->setDefaultTopicConf($topicConf);
    //
    //     $consumer = new \RdKafka\KafkaConsumer($conf);
    //     $consumer->subscribe(["shop"]);
    //     while (true) {
    //         $message = $consumer->consume(1000);
    //         switch ($message->err) {
    //             case RD_KAFKA_RESP_ERR_NO_ERROR:
    //                 echo "正确读取到数据\n\r";
    //                 var_dump($message->payload);
    //                 break;
    //             case RD_KAFKA_RESP_ERR__PARTITION_EOF:
    //                 echo "没有读到数据\n\r";
    //                 break;
    //             case RD_KAFKA_RESP_ERR__TIMED_OUT:
    //                 echo "读取数据超时\n\r";
    //                 break;
    //             default:
    //                 throw new \Exception($message->errstr(), $message->err);
    //                 break;
    //         }
    //     }
    //
    //     $output->writeln("!!!the end!!!");
    // }


}